# multiprocessing 多进程模块
# multiprocessing模块支持fork子进程执行任务
# multiprocessing模块用法与threading相似, 基本上threading使用方法照搬过来就可以
# 包括threading中的lock/rlock/semaphore等，除此之外还有一些进程独有的方法如getpid
# exitcode/terminate等
# 使用子进程方式可以避免Python GIL机制对于多线程并发执行的限制

# Process子进程使用
# 创建子进程根据操作系统不同，主要有三种方式：
#   1. spawn 直接创建新进程，将由python解释器直接启动一个新的进程
#      该进程只继承父进程run方法所需的函数或者资源
#   2. fork 父进程使用os.fork()产生一个进程，该进程完全继承父进程所有资源
#      fork产生的进程和父进程完全一致，注意：多线程os.fork()进程会出现安全性问题
#   3. forkserver 程序启动时启动一个forkserver进程，当程序需要子进程时连接到forkserver进程
#      请求创建一个新进程，因为forserver进程是单线程所以使用os.fork()是安全的，不会继承不必要的资源
from multiprocessing import Process
import os # 用于获取进程信息

def f1(name):
    print(f'hello {name}')

def pinfo():
    print('module name->', __name__)
    print('parent process->', os.getppid(), 'current porcess->', os.getpid())

def processMain():
    pinfo()
    p = Process(target=f1, args=('tom',)) # 创建进程并只指定参数 使用方式类似threading模块线程
    p.start() # 启动进程
    p.join() # jion到主进程 阻止主进程退出

def f2(q):
    q.put('hello')


# 进程间传递数据方式有两种：一种Queue，一种Pipe
def forkProcessMain():
    import multiprocessing as mp
    mp.set_start_method('spawn') # 指定fork子进程的方式, 只能调用一次
    q = mp.Queue() # 创建一个队列
    p = mp.Process(target=f2, args=(q,)) # 创建进程，指定执行方法
    p.start()
    print('from subprocess value ->', q.get())
    p.join() # 

def processContextExchangeDataMain():
    ''' context关联的对象可能在不同的context见不兼容，
        比如：使用fork方式创建的锁不能传递到spawn和forkserver方式启动的进程中
     '''
    import multiprocessing as mp
    ctx = mp.getContext('spawn') # 获取上下文
    q = ctx.Queue() # 创建一个队列
    p = ctx.Process(target=f2, args=(q,)) # 创建进程，指定执行方法
    p.start()
    print('ctx from subprocess value ->', q.get())
    p.join() # 

def f3(conn):
    conn.send(['hello',123])
    print('child recv->', conn.recv())
    conn.close()

def pipeExchangeDataMain():
    from multiprocessing import Pipe # 导入包
    parent_conn, child_conn = Pipe() # 返回两个PipeConnection 对象分别代表管道的两头
    # help(parent_conn) # PipeConnection
    # help(child_conn)
    p = Process(target=f3, args=(child_conn,))
    p.start()
    parent_conn.send('hi child')
    print('parent recv ->', parent_conn.recv()) # 接收数据
    parent_conn.close()
    p.join()

def f4(lock, i):
    with lock: # 哪个进程获取到锁哪个就可以执行
        import time
        time.sleep(0.2)
        print('hi ->', i)

# 进程同步，multiprocessing包含threading所有的同步原语
def processSync():
    from multiprocessing import Lock
    lock = Lock()
    for i in range(5): # 启动5个进程，同一时间指挥有一个输出
        Process(target=f4, args=(lock, i)).start()

def f5(n, a):
    n.value=3.14159
    for i in range(len(a)):
        a[i] = -a[i]

# 通常情况不建议进程间共享数据
# 进程间共享数据方式：
#   1. sharedMemeory共享内存Value Array对象，更灵活的可以使用multiprocessing.sharedctypes
#   2. server process manager 进程管理对象Manager管理python进程，允许通过该对象操作数据
#      支持dict、list、lock、rlock、queue、value等，还可以跨机器共享相比共享内存灵活但是速度慢一些
def processShareMemory():
    from multiprocessing import Value, Array
    n = Value('d', 0.0) # 创建double精度的value对象
    ar = Array('i', range(10)) # 创建int类型10个数的数组
    p = Process(target=f5, args=(n,ar))
    p.start()
    p.join()
    print(f'shared value->{n.value}')
    print(f'shared array->{ar[:]}') # ar是个array对象，通过ar[:]取出所有数据

def f6(d, l):
    ''' 字典赋值，列表取反 '''
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()
# 此外还支持remote manager远程数据共享QueueManager,参考Managers
def processManagerShareData():
    from multiprocessing import Manager
    with Manager() as m:
        d = m.dict() # 获取的是dict的代理
        l = m.list(range(10))
        p = Process(target=f6, args=(d,l))
        p.start()
        p.join()
        print('manger -> ', d, l)

# Pool对象通过创建worker进程，可以使函数并行执行
from multiprocessing import Pool
def f7(x):
    return x * x
# pool对象使用方式
def poolMain():
    # 创建4个工作进程, with 内代码块执行结束自动close关闭
    with Pool(4) as p:
        # 执行f函数将列表的每个值作为参数进行并行计算
        print(p.map(f7, [1,2,3]))

        # 随机计算输出
        for i in p.imap_unordered(f7, range(10)):
            print(i)

        # 异步执行f7(20)函数， 只会在一个进程中执行
        res = p.apply_async(f7, (20,))
        print('async ->', res.get(timeout=1)) # 获取结果

        # 异步执行os.getpid()方法
        print('async getpid ->', p.apply_async(os.getpid, ()).get(timeout=1))

        # 异步执行多个任务，多个进程
        multi_res = [p.apply_async(os.getpid,()) for i in range(4)]
        print([r.get(timeout=1) for r in multi_res]) # 输出结果

        import time
        from multiprocessing import TimeoutError
        # 异步睡眠1秒, 模拟程序执行3秒
        res = p.apply_async(time.sleep, (3,))
        try:
            print(res.get(timeout=1)) # 超时时间1秒，执行大于1秒抛出timeoutError
        except TimeoutError as e:
            print('timeout async exec ->', e)

# listener client是Pipe的高层封装，使用更灵活简便
from multiprocessing.connection import Listener, Client
from array import array
addr = ('localhost', 6666)
def conn_listener():
    with Listener(addr, authkey=b'pass') as listener: # 监听并带有密码认证
        with listener.accept() as conn:
            print('client connect from ', listener.last_accepted)
            conn.send([2,None,'tom','cat'])
            conn.send_bytes(b'hello')
            print(conn.send_bytes(array('i', range(3,5))))

def conn_client():
    with Client(addr, authkey=b'pass') as conn:
        print('client recv ->', conn.recv())
        print('client recv bytes ->', conn.recv_bytes())
        arr = array('i', [0,0,0,0]) # into的数组要大于等于发过来的数组长度
        print(conn.recv_bytes_into(arr)) # 收到8个字节
        print('client recv bytes into ->' , arr)

def listen_client():
    p = Pool(2)
    try:
        # 异步执行，启动两个进程
        p.apply_async(conn_listener, ())
        p.apply_async(conn_client, ())
    finally:
        p.close() # 关闭，禁止提交新任务，并且等待所有worker执行完毕然后退出
        # p.terminate() # 停止所有执行中的任务 马上退出
        p.join() # 必须在close或者terminate之后调用，等待所有worker退出


import time,random
from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait
def write_i(w):
    for i in range(3):
        w.send((i, current_process().name)) # 发送元组
    w.close()

def waitReadMulitProcess():
    readers = []
    for i in range(4):
        r,w = Pipe(duplex=False) # 双向 false
        readers.append(r)
        p = Process(target=write_i, args=(w,))
        p.start()
        # 关闭本进程写入句柄以确保只有p进程拥有该写入端句柄(子进程继承父进程资源)
        # 该句柄关闭后，就只有子进程有一个写入句柄(相当于父进程一个副本，不过父进程关闭了也就不再拥有了)
        # 当子进程调用w.close()时会触发wait条件通知r可读就绪
        w.close() 

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv() # 接收数据
            except EOFError as e: # 子进程中close时引发eof异常
                readers.remove(r)
            else:
                print('read ->', msg)

# 进程间日志不使用共享锁
def multiprocessingLog():
    import multiprocessing, logging
    # log = multiprocessing.get_logger() # 需要配置handler formatter
    log = multiprocessing.log_to_stderr()
    log.setLevel(logging.INFO)
    log.warning('ok')
    m = multiprocessing.Manager() # 创建manager会打印日志
    del m # 删除也会打印

# 多进程编程注意事项
#   1. 避免多进程间共享数据，如果需要推荐使用Queue或Pipe，而不是使用同步原语(lock/barrier..)
#   2. 多线程如果使用proxy代理对象，需要加锁，多进程见不需要
#   3. join僵尸进程，有时候已经完成的进程还为join会变成zombie僵尸进程，
#      对于已完成的进程调用process.is_alive将会join进程。最好是启动后close再调用join
#   4. 进程间的数据最好将需要的数据继承到子进程好于同各国Pipe或queue传过去
#   5. 在进程间存在共享数据时，进程不要调用terminate，会导致共享数据出现异常
#   6. 使用queue传递数据时，queue数据获取完毕再join
#   7. 子进程target函数使用的数据要通过参数传递
#   8. 进程要定义main入口  __name__ == '__main__' 不同的平台上启动进程需要入口方法
if __name__ == '__main__':
    # processMain()
    # forkProcessMain()
    # processContextExchangeDataMain()
    # pipeExchangeDataMain()
    # processSync()
    # processShareMemory()
    # processManagerShareData()
    # poolMain()
    # listen_client()
    # waitReadMulitProcess()
    multiprocessingLog()







